package com.li.controller.listener;

import com.li.controller.TestController;
import com.li.filter.RocketMqConsumerMessageHook;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.stereotype.Component;

/**
 * @program: srpingboot-rocketmq
 * @description:
 * @author: li
 * @create: 2022-09-23 14:19
 **/
@Component
@RocketMQMessageListener(
        consumerGroup = "common_product_group_name",
//        messageModel = MessageModel.BROADCASTING,
        topic = "TOPIC_TEST")
/**
 * messageModel  设置消息消费 模式， 默认 CLUSTERING 集群模式, BROADCASTING 广播模式
 */
public class CommonMessageConsumer implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {
    private static final Logger logger = LoggerFactory.getLogger(CommonMessageConsumer.class);

    @Override
    public void onMessage(MessageExt message) {
//        logger.info("CommonMessageConsumer common_product_group_name_2 收到消息：{}", new String(message.getBody()));
        logger.info("CommonMessageConsumer common_product_group_name 收到消息：{}", new String(message.getBody()));
        // 业务逻辑, 默认没有异常即消费成功， 返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS
        System.out.println("traceid" + message.getUserProperty("traceid"));
        MDC.put("traceId", message.getUserProperty("traceid"));
        System.out.println("当前消息第" + message.getReconsumeTimes() + "次重试" );
        logger.info("LOGBACK - MDC:" + MDC.get("traceId"));

//        // 模拟 多次消费失败后，不再重复消费
//        if (message.getReconsumeTimes() == 3) {
//            // insert into log
//            return;
//        }
//        System.out.println("----------接收到rocketmq消息:" + new String(message.getBody()));
//        // rocketmq会自动捕获异常回滚  (官方默认会重复消费16次， 默认 30s 后重新消费)

//        int a = 1 / 0;


    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        // 给当前消费者设置相关属性
        // CONSUME_FROM_LAST_OFFSET  默认策略，从该队列最尾部开始消费， 跳过历史消息
        // CONSUME_FROM_FIRST_OFFSET 从队列最开始的消息 开始消费 即历史消息（还储存在broker的）全部消费一遍
        // CONSUME_FROM_TIMESTAMP  从某个指定时间节点开始消费
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // 设置最大重试次数
//        defaultMQPushConsumer.setMaxReconsumeTimes(20);
        defaultMQPushConsumer.setMaxReconsumeTimes(1);
        // 设置钩子方法
        defaultMQPushConsumer.getDefaultMQPushConsumerImpl().registerConsumeMessageHook(new RocketMqConsumerMessageHook());

    }
}
